/*
 * Project: push-center-service-impl
 * 
 * File Created at 2014-10-13
 * 
 * Copyright 2012 Greenline.com Corporation Limited.
 * All rights reserved.
 *
 * This software is the confidential and proprietary information of
 * Greenline Company. ("Confidential Information").  You shall not
 * disclose such Confidential Information and shall use it only in
 * accordance with the terms of the license agreement you entered into
 * with Greenline.com.
 */
package com.greenline.push.service.core.send.base;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.TimeUnit;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;

import org.apache.commons.lang.StringUtils;
import org.apache.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;

import com.alibaba.fastjson.JSON;
import com.greenline.push.core.enums.Tasks;
import com.greenline.push.core.pushcenter.GroupUserManager;
import com.greenline.push.core.pushcenter.MsgRetryFullManager;
import com.greenline.push.core.pushcenter.MsgRetryManager;
import com.greenline.push.core.pushcenter.TaskSyncManager;
import com.greenline.push.dal.pushcenter.dataobject.GroupUserDO;
import com.greenline.push.dal.pushcenter.dataobject.MsgRetryDO;
import com.greenline.push.dal.pushcenter.dataobject.TaskSyncDO;
import com.greenline.push.service.core.ProcessQueueList;
import com.greenline.push.service.core.counter.CounterManager;
import com.greenline.push.service.core.exception.ChannelSendException;
import com.greenline.push.service.core.exception.TaskException;
import com.greenline.push.service.core.item.AppPacket;
import com.greenline.push.service.core.item.Packet;
import com.greenline.push.service.core.task.FetchRetryDataSchedule;
import com.greenline.push.service.core.task.FetchRetryFullDataSchedule;
import com.greenline.push.service.core.tmpl.AppTemplateInfo;
import com.greenline.push.service.core.user.AppUserInfo;
import com.greenline.push.service.core.user.UserConnact;
import com.greenline.push.service.core.user.UserManager;
import com.greenline.push.service.core.utils.Constants;

/**
 * AbstractSender
 * 
 * @author weirui.shenwr
 * @date 2014-10-13
 * @version V1.0
 */
public abstract class AbstractSender implements ISender {

    Logger log = Constants.LOG_PROCESS;
    @Resource
    private UserManager userManager;
    @Autowired
    private TaskSyncManager taskSyncManager;

    private final int sending_queue_count = 3;
    private final int sending_queue_listener_count = 3;
    private int maxRemainingCapacity = 1200;

    private boolean isStart = false;
    private boolean isStartFull = false;
    @Resource
    private CounterManager conterManager;
    @Resource
    private MsgRetryManager msgRetryManager;
    @Resource
    private MsgRetryFullManager msgRetryFullManager;
    @Resource
    private FetchRetryDataSchedule fetchRetrySchedule;
    @Resource
    private FetchRetryFullDataSchedule fetchRetryFullSchedule;
    @Resource
    private AppPacketBuilder appPacketBuilder;
    @Resource
    private GroupUserManager groupUserManager;

    private List<SendingQueueListener> threads = null;
    private List<SendingQueueListener> fullthreads = null;

    /**
     * 待发送的队列
     */
    private ProcessQueueList sendingQueue;
    /**
     * 全量发送队列
     */
    private ProcessQueueList sendingQueueFull;

    @PostConstruct
    public void start() throws TaskException {
        // 普通推送任务处理器
        startDefaultProcessor();
        // 全量推送任务处理器
        startFullProcessor();
    }

    private void startDefaultProcessor() throws TaskException {
        if (!isStart) {
            log.info(String.format("%sSender初始化开始.本次启动唯一应用id为:%s", getQueueName(), Constants.APP_KEY));

            // 初始化队列
            sendingQueue = new ProcessQueueList(sending_queue_count, maxRemainingCapacity);

            // 启动重发任务
            // index_1:队列 index_2:任务类型 index_3:任务执行间隔(s) 4.时间类型
            TaskSyncDO task = taskSyncManager.getTaskSyncByType(getTaskType().getType());
            if (null == task) {
                log.error(String.format("%s.该任务数据库未配置.会导致该发送通道不通.请及时检查.", getTaskType()));
                throw new NullPointerException(String.format("%s.该任务数据库未配置.会导致该发送通道不通.请及时检查.", getTaskType()));
            }
            fetchRetrySchedule.startTask(sendingQueue, getTaskType(), supportChannel(), task.getRunPeriod(), TimeUnit.SECONDS);

            // 加入计数器
            conterManager.addQueue(getQueueName(), sendingQueue);

            // 启动线程
            threads = new ArrayList<SendingQueueListener>(sending_queue_count * sending_queue_listener_count);
            for (int i = 0; i < sending_queue_count; i++) {
                for (int j = 0; j < sending_queue_listener_count; j++) {
                    // start sender p2p and group
                    SendingQueueListener listener = new SendingQueueListener(i, sendingQueue);
                    listener.setName("[消息发送处理器][队列:" + getTaskType() + (i * sending_queue_listener_count + j));
                    listener.start();
                    threads.add(listener);
                }
            }
            isStart = true;
            log.info(String.format("%sSender初始化完成.本次启动唯一应用id为:%s", getQueueName(), Constants.APP_KEY));
        }
    }

    private void startFullProcessor() throws TaskException {
        if (!isStartFull) {
            Tasks tasks = getFullTaskType();
            log.info(String.format("%sSender初始化开始.本次启动唯一应用id为:%s", getFullQueueName(), Constants.APP_KEY));
            // 初始化全量发送队列
            sendingQueueFull = new ProcessQueueList(sending_queue_count, maxRemainingCapacity);
            // 全量重发任务
            fetchRetryFullSchedule.startTask(sendingQueueFull, tasks, supportChannel(), taskSyncManager.getTaskSyncByType(tasks.getType())
                    .getRunPeriod(), TimeUnit.SECONDS);
            // 加入计数器
            conterManager.addQueue(getFullQueueName(), sendingQueueFull);
            // 启动线程
            fullthreads = new ArrayList<SendingQueueListener>(sending_queue_count * sending_queue_listener_count);
            for (int i = 0; i < sending_queue_count; i++) {
                for (int j = 0; j < sending_queue_listener_count; j++) {
                    // start sender p2p and group
                    SendingQueueListener listener = new SendingQueueListener(i, sendingQueueFull);
                    listener.setName("[全量消息发送处理器][队列:" + tasks + (i * sending_queue_listener_count + j));
                    listener.start();
                    fullthreads.add(listener);
                }
            }
            isStartFull = true;
            log.info(String.format("%sSender初始化完成.本次启动唯一应用id为:%s", getFullQueueName(), Constants.APP_KEY));
        }
    }

    /**
     * 获得用户的联系方式
     * 
     * @param userId
     * @return
     */
    protected UserConnact getUserConnact(Long userId) {
        return userManager.getUserConnact(userId);
    }

    /**
     * weixin 获取openid 或者 其他的token
     * 
     * @param userId
     * @param appId
     * @return
     */
    protected AppUserInfo getAppUser(Long userId, String appId) {
        AppUserInfo appUser = userManager.getAppUserInfo(userId, appId);
        return appUser;
    }

    protected AppUserInfo getAppUser(String userId, String appId) {
        if (StringUtils.isNumeric(userId)) {
            return this.getAppUser(Long.valueOf(userId), appId);
        }
        return null;
    }

    /**
     * true if insert queue succ,false if insert queue fail;
     */
    @Override
    public boolean send(Packet packet, AppTemplateInfo appTempInfo) {
        AppPacket p = appPacketBuilder.buildChannelPacket(packet, appTempInfo, supportChannel());
        if (p == null) {
            return false;
        }
        // first insert queue and get id
        MsgRetryDO msgRetryDo = insertMsgRetry(p, appTempInfo, true);
        //
        p.setDbRetryId(msgRetryDo.getId());
        boolean t = sendingQueue.offer(p);
        // if insert to queue fail,update db
        if (!t) {
            MsgRetryDO m = new MsgRetryDO();
            m.setId(msgRetryDo.getId());
            m.setFirstOfferStatus(1);
            m.setMemo("阀值超过,放入send_try表重试。");
            updateMsgRetry(m);
        }
        // will return true weather or not insert queue success or not.
        return true;
    }

    /**
     * 调用真正的发送
     * 
     * @param packet
     * @param appTempInfo
     * @return
     */
    protected abstract boolean doSend(Packet packet);

    /**
     * 获取普通任务类型
     * 
     * @return
     */
    protected abstract Tasks getTaskType();

    /**
     * 获取全量任务类型
     * 
     * @return
     */
    protected abstract Tasks getFullTaskType();

    // protected abstract boolean doSend(List<Packet> packetList);

    /**
     * 获取普通队列名
     * 
     * @return
     */
    protected abstract String getQueueName();

    /**
     * 获取全量队列名
     * 
     * @return
     */
    protected abstract String getFullQueueName();

    /**
     * 
     * ProcessQueueListener
     * 
     * @author weirui.shenwr
     * @date 2014-10-10
     * @version V1.0
     */
    private class SendingQueueListener extends Thread {

        private int queueIndex;
        private boolean threadStopped = false;
        ProcessQueueList queueList;

        public SendingQueueListener(int queueIndex, ProcessQueueList queueList) {
            this.queueIndex = queueIndex;
            this.queueList = queueList;
        }

        @Override
        public void run() {
            while (!threadStopped) {

                Packet packet = null;
                try {
                    packet = queueList.take(queueIndex);
                    // 调用渠道推送处理
                    doSend(packet);
                } catch (Exception e) {
                    log.error("[" + getName() + "] Exception during packet processing: " + packet, e);
                }
            }
        }

        public void threadStop() {
            threadStopped = true;
        }
    }

    /**
     * 插入重发日志
     * 
     * @author zhaoli
     * @param packet
     * @param appTempInfo
     * @param isSucc 是否超时
     * @throws ChannelSendException
     */
    private MsgRetryDO insertMsgRetry(AppPacket packet, AppTemplateInfo appTempInfo, boolean isSucc) {
        MsgRetryDO msgRetryDO = new MsgRetryDO();
        msgRetryDO.setAppChannel(this.supportChannel().getType());
        msgRetryDO.setAppId(appTempInfo.getAppId());
        if (isSucc) {
            msgRetryDO.setFailCount(0);
            msgRetryDO.setMemo("放人队列成功。");
            msgRetryDO.setFirstOfferStatus(0);
        } else {
            msgRetryDO.setFailCount(0);
            msgRetryDO.setMemo("阀值超过,放入send_try表重试。");
            msgRetryDO.setFirstOfferStatus(1);
        }
        msgRetryDO.setPacketJson(JSON.toJSONString(packet));
        msgRetryDO.setTemplateId(appTempInfo.getTemplateId());
        msgRetryManager.insert(msgRetryDO);
        return msgRetryDO;
    }

    private boolean updateMsgRetry(MsgRetryDO msgretry) {
        if (msgretry.getId() == null) {
            return false;
        }
        return msgRetryManager.update(msgretry);
    }

    /**
     * 群发情况下 ,根据群 id获取所有订阅用户
     * 
     * @param gid
     * @return
     */
    protected List<Long> getUserIdsbyGid(Long gid) {
        List<Long> userIds = new ArrayList<Long>();
        List<GroupUserDO> sList = groupUserManager.queryByGid(gid);
        for (GroupUserDO groupUserDO : sList) {
            userIds.add(groupUserDO.getUserId());
        }
        return userIds;
    }

    /**
     * send list 调用时 ,获取msg_retry_full 表中的数据
     * 
     * @return
     */
    protected List<String> getUserIdsByRetryId(Long retryId) {
        MsgRetryDO retry = msgRetryFullManager.getById(retryId);
        if (null == retry) {
            log.error(String.format("{getUserIdsByRetryId} msg_retry_full 下id:%s的数据不存在", retryId));
            return null;
        }
        String uidString = retry.getUidList();
        if (StringUtils.isBlank(uidString)) {
            // log.error(String.format("{getUserIdsByRetryId}msg_retry_full 下uid_list 为空. 数据删除",
            // retryId));
            // msgRetryFullManager.delete(retryId);
            return null;
        }
        List<String> sResult = Arrays.asList(uidString.split(","));
        return sResult;
    }
}
